package com.bbx.flink.demo.data_stream_state.functions;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.Map;

/**
 * 重置Ttl 时间  的策略
 * StateTtlConfig.UpdateType.OnCreateAndWrite       state 创建和 写入时，Ttl 重新计时
 *  StateTtlConfig.UpdateType.OnReadAndWrite          state 写入或者读取时 都会重新计时
 *
 *
 *
 */
@Slf4j
public class TtlStateDemo extends RichFlatMapFunction<Integer, Map<String ,Integer>> {
    private static final long serialVersionUID = 1696243549837165398L;

    private ValueState<Integer> valueState;



    @Override
    public void open(Configuration parameters) throws Exception {
        StateTtlConfig ttlConfig = StateTtlConfig.newBuilder(Time.milliseconds(8000))
                //state 创建和 写入时，Ttl 重新计时
                //setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                // 写入或者读取时 都会重新计时
                .setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite)
                //state 过期 返回过期的用户数据--大概明白，但是测试没有效果，应该要结合过期 cleanUp 策略
//                .setStateVisibility(StateTtlConfig.StateVisibility.ReturnExpiredIfNotCleanedUp)
                //state 过期后不返回过期的用户数据 --大概明白，但是测试没有效果，应该要结合过期 cleanUp 策略
                .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                .build();

        ValueStateDescriptor<Integer> stateDescriptor = new ValueStateDescriptor<>("valueState", Integer.class);
        stateDescriptor.enableTimeToLive(ttlConfig);
        valueState =  getRuntimeContext().getState(stateDescriptor);




    }

    @Override
    public void flatMap(Integer value, Collector<Map<String, Integer>> out) throws Exception {
        Integer oldValue = valueState.value();
        log.info("oldValue[{}]",oldValue);

        if (oldValue != null) {
            int abs = Math.abs(value - oldValue);
            if(abs>5){
                HashMap<String, Integer> map = new HashMap<>();
                map.put("oldValue" ,oldValue);
                map.put("currentValue", value);
                out.collect(map);
            }
        }
        if(value==1){
            //配合验证    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
            valueState.update(value);
        }

    }
}
